package com.itheima.es;

import com.alibaba.fastjson.JSON;
import com.itheima.es.pojo.Goods;
import com.itheima.es.pojo.User;
import com.itheima.es.service.UserService;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;

import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.Avg;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ElasticDemo {

    private RestHighLevelClient client;

    /**
     * 建立连接
     */
    @Before
    public void init() throws IOException {
        client = new RestHighLevelClient(
                RestClient.builder(
                  //      new HttpHost("localhost", 9200, "http")
                        new HttpHost("192.168.206.99", 9200, "http")
                )
        );
    }

    /**
     * 关闭客户端连接
     */
    @After
    public void close() throws IOException {
        client.close();
    }

    // 测试
    @Test
    public void testCreateIndexes() throws IOException {

        // 1.创建索引库
        CreateIndexRequest indexRequest = new CreateIndexRequest("user");
        // 2.指定settings配置
        indexRequest.settings(Settings.builder()
                .put("index.number_of_shards", 3)
                .put("index.number_of_replicas", 1)
        );
        // 3.指定mapping配置
        indexRequest.mapping("{\n" +
                        "    \"properties\": {\n" +
                        "      \"id\": {\n" +
                        "        \"type\": \"long\"\n" +
                        "      },\n" +
                        "      \"name\":{\n" +
                        "        \"type\": \"keyword\"\n" +
                        "      },\n" +
                        "      \"age\":{\n" +
                        "        \"type\": \"integer\"\n" +
                        "      },\n" +
                        "      \"gender\":{\n" +
                        "        \"type\": \"keyword\"\n" +
                        "      },\n" +
                        "      \"note\":{\n" +
                        "        \"type\": \"text\",\n" +
                        "        \"analyzer\": \"ik_max_word\"\n" +
                        "      }\n" +
                        "    }\n" +
                        "  }",
                XContentType.JSON);
        //  发起请求得到响应
        CreateIndexResponse response = this.client.indices().create(indexRequest, RequestOptions.DEFAULT);

        System.out.println("response.isAcknowledged() = " + response.isAcknowledged());
        System.out.println("response.isShardsAcknowledged() = " + response.isShardsAcknowledged());

    }

    // 根据索引名删除库
    @Test
    public void testDeleteIndexes() throws IOException {

        DeleteIndexRequest deleteRequest = new DeleteIndexRequest("user");
        AcknowledgedResponse acknowledgedResponse = this.client.indices().delete(deleteRequest, RequestOptions.DEFAULT);

        System.out.println("acknowledgedResponse.isAcknowledged() = " + acknowledgedResponse.isAcknowledged());
    }

    //  手动数据导入
    @Test
    public void testAddDocument() throws IOException {

        // 创建IndexRequest对象，指定索引库名称
        IndexRequest indexRequest = new IndexRequest("user");
        //  id在lucene里面数值存储会转义
        //  例:123————口口口  似MD5加密后的明文和密文  存入是123，查询的时候结果是不相等所以查询不到
        // 在es中无论给任何类型的数据，存储方式都是按string类型
        indexRequest.id("10086");
        // source是数据真正存放的地方，其数据结构是JSON请求
        indexRequest.source("{\n" +
                "  \"id\": 10086,\n" +
                "  \"name\": \"上海肛肠医院何院长\",\n" +
                "  \"gander\": \"男\",\n" +
                "  \"age\": 18,\n" +
                "  \"note\": \"正在给波多野结衣检查身体\"\n" +
                "}", XContentType.JSON);
        // 请求并响应
        IndexResponse indexResponse = this.client.index(indexRequest, RequestOptions.DEFAULT);
        System.out.println("indexResponse.status() = " + indexResponse.status());
    }


    //  新增数据手动从数据库查询文档数据

    // 1.在测试类中初始化UserService对象
    public UserService userService = new UserService();

    @Test
    public void testAddDocumentFromDB() throws IOException {
        // 指定新增的数据id
        User user = this.userService.findById(6L);
        // 查看相关工具类
       // StringUtils.toString()

        // 将新增文档数据转成JSON格式
        // user序列化后
        String userjson = JSON.toJSONString(user);
        // 创建IndexRequest对象根据索引库名称查询
        IndexRequest indexRequest = new IndexRequest("user");
        indexRequest.id(String.valueOf(user.getId()));   // toString易造成空指针
        // 将json数据加入到IndexRequest中
        indexRequest.source(userjson, XContentType.JSON);

        // 发起请求并得到结果
        IndexResponse indexResponse = this.client.index(indexRequest, RequestOptions.DEFAULT);
        System.out.println("indexResponse.status() = " + indexResponse.status());
    }

    // 批量导入数据

    @Test
    public void testAddDocumentFromDBBatch() throws IOException {
        // 从数据库查询文档数据
        List<User> userList = userService.findAll();
        // 创建BulkRequest对象
        BulkRequest bulkRequest = new BulkRequest("");
        // 创建多个bulkRequest对象
        for (User user : userList) {
            bulkRequest.add(new IndexRequest("user").id(user.getId().toString()).source(JSON.toJSONString(user),XContentType.JSON));
        }
        // 请求并响应
        BulkResponse bulkresponse = this.client.bulk(bulkRequest, RequestOptions.DEFAULT);
        System.out.println("bulkresponse.status() = " + bulkresponse.status());
    }

    //  根据id查询
    @Test
    public void testQueryById() throws IOException {
        // 索引库为user根据指定id查询
        GetRequest getRequest = new GetRequest("user","10086");
        // 发送请求并得到响应
        GetResponse getResponse = this.client.get(getRequest, RequestOptions.DEFAULT);
        // 从结果中获取source，自动将JSON转换为string
        String userjson = getResponse.getSourceAsString();
        // 反序列化后转为对象
        User user = JSON.parseObject(userjson, User.class);
        System.out.println("user = " + user);
    }



    @Test
    public void testQuery() throws IOException {

        SearchRequest searchRequest = new SearchRequest("user");

        // 条件构建工具:查询、分页、高亮、聚合、排序、字段过滤
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 在条件工具中添加查询条件,分词查询
       // sourceBuilder.query(QueryBuilders.matchQuery("note","传智播客"));
          sourceBuilder.query(QueryBuilders.termQuery("note","java"));

          // 分页
          sourceBuilder.from(3);
          sourceBuilder.size(3);

  /*      HighlightBuilder highlightBuilder = new HighlightBuilder();

        highlightBuilder.preTags("<em>");
        highlightBuilder.postTags("</em>");
        highlightBuilder.field("note");*/

//        // 添加高亮显示
//        sourceBuilder.highlighter(highlightBuilder);

        // 添加条件
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = this.client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits searchHits = searchResponse.getHits();

        // 命中数量
        long totaHits = searchHits.getTotalHits().value;
        System.out.println("查询命中数为:" + totaHits);
        // 实际查询到的元素集合
        SearchHit[] hits = searchHits.getHits();

        for (SearchHit hit : hits) {
            float score = hit.getScore();
            User user = JSON.parseObject(hit.getSourceAsString(), User.class);
            System.out.println("得分:" + score + "info:" + user);
        }
    }

    // 添加高亮
    @Test
    public void testQueryHighLight() throws IOException {

        SearchRequest searchRequest = new SearchRequest("user");

        // 条件构建工具:查询、分页、高亮、聚合、排序、字段过滤
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 在条件工具中添加查询条件,分词查询
        sourceBuilder.query(QueryBuilders.matchQuery("note","传智播客"));


        HighlightBuilder highlightBuilder = new HighlightBuilder();

        highlightBuilder.preTags("<em>");
        highlightBuilder.postTags("</em>");
        highlightBuilder.field("note");

        // 添加高亮条件
        sourceBuilder.highlighter(highlightBuilder);

        // 添加条件
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = this.client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits searchHits = searchResponse.getHits();

        // 命中数量
        long totaHits = searchHits.getTotalHits().value;
        System.out.println("查询命中数为:" + totaHits);
        // 实际查询到的元素集合
        SearchHit[] hits = searchHits.getHits();

        for (SearchHit hit : hits) {
            float score = hit.getScore();
            User user = JSON.parseObject(hit.getSourceAsString(), User.class);
            // 从查询结果中获取高亮信息
            Text note = hit.getHighlightFields().get("note").fragments()[0];
            // 将高亮信息String化处理
            String joinSt = StringUtils.join(note);
            // 将高亮信息赋值给具体的高亮属性
            user.setNote(joinSt);
            System.out.println("得分:" + score + "info:" + user);
        }
    }


    // 聚合

    public void testAggs() throws IOException {
        // 查询请求对象
        SearchRequest searchRequest = new SearchRequest("car");
        //  条件构建工具
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        //  不展示查询结果
        sourceBuilder.size(0);
        //  添加聚合,词条内容聚合
        sourceBuilder.aggregation(AggregationBuilders.terms("makes").field("make").size(10)
                .subAggregation(AggregationBuilders.avg("avg_price").field("price")));

        //  添加条件
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = this.client.search(searchRequest, RequestOptions.DEFAULT);
        //  获取所有的聚合
        Aggregations aggregations = searchResponse.getAggregations();
        // 根据聚合名称获取对应的聚合
        Terms makes = aggregations.get("makes");
        // 获取到所有的聚合分桶
        List<? extends Terms.Bucket> buckets = makes.getBuckets();
        buckets.forEach(bucket->{
            //  由于参与聚合的key类型比较多,所有事object
            String key = bucket.getKeyAsString();
            long docCount = bucket.getDocCount();
            Avg avg = bucket.getAggregations().get("avg_price");
            double avg_price = avg.getValue();
            System.out.println("key" + "有" + docCount +"个,平均价格为 :"+ avg_price);
        });
    }

    /**
     *    创建索引库(Kibana)
     * @throws IOException
     */
    @Test
        public void testCreateIndexBo() throws IOException {
            CreateIndexRequest RequestCreateIndexBo = new CreateIndexRequest("goods");

            RequestCreateIndexBo.source("{\n" +
                    "  \"settings\": {\n" +
                    "    \"analysis\": {\n" +
                    "      \"analyzer\": {\n" +
                    "        \"my_pinyin\": {\n" +
                    "          \"tokenizer\": \"ik_smart\",\n" +
                    "          \"filter\": [\n" +
                    "            \"py\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      },\n" +
                    "      \"filter\": {\n" +
                    "        \"py\": {\n" +
                    "          \"type\": \"pinyin\",\n" +
                    "          \"keep_full_pinyin\": false,\n" +
                    "          \"keep_joined_full_pinyin\": true,\n" +
                    "          \"keep_original\": true,\n" +
                    "          \"limit_first_letter_length\": 16,\n" +
                    "          \"remove_duplicated_term\": true\n" +
                    "        }\n" +
                    "      }\n" +
                    "    }\n" +
                    "  },\n" +
                    "  \"mappings\": {\n" +
                    "    \"properties\": {\n" +
                    "      \"id\": {\n" +
                    "        \"type\": \"keyword\"\n" +
                    "      },\n" +
                    "      \"name\": {\n" +
                    "        \"type\": \"completion\",\n" +
                    "        \"analyzer\": \"my_pinyin\",\n" +
                    "        \"search_analyzer\": \"ik_smart\"\n" +
                    "      },\n" +
                    "      \"title\":{\n" +
                    "        \"type\": \"text\",\n" +
                    "        \"analyzer\": \"my_pinyin\",\n" +
                    "        \"search_analyzer\": \"ik_smart\"\n" +
                    "      },\n" +
                    "      \"price\":{\n" +
                    "        \"type\": \"long\"\n" +
                    "      }\n" +
                    "    }\n" +
                    "  }\n" +
                    "}",XContentType.JSON);

            CreateIndexResponse createIndexResponse = this.client.indices().create(RequestCreateIndexBo, RequestOptions.DEFAULT);

            System.out.println("createIndexResponse.isAcknowledged() = " + createIndexResponse.isAcknowledged());
        }

    /**
     *    批量添加数据到索引库中
      * @throws IOException
     */
    @Test
    public void testBulkDocument() throws IOException {
        // 1.准备文档数据
        List<Goods> list = new ArrayList<>();
        list.add(new Goods(1L, "红米9", "红米9手机 数码", 1499L));
        list.add(new Goods(2L, "三星 Galaxy A90", "三星 Galaxy A90 手机 数码 疾速5G 骁龙855", 3099L));
        list.add(new Goods(3L, "Sony WH-1000XM3", "Sony WH-1000XM3 降噪耳机 数码", 2299L));
        list.add(new Goods(4L, "松下剃须刀", "松下电动剃须刀高转速磁悬浮马达", 599L));
        // 2.创建BulkRequest对象
        BulkRequest bulkRequest = new BulkRequest();
        // 3.创建多个IndexRequest对象，并添加到BulkRequest中
        for (Goods goods : list) {
            bulkRequest.add(new IndexRequest("goods")
                    .id(goods.getId().toString())
                    .source(JSON.toJSONString(goods), XContentType.JSON)
            );
        }
        // 4.发起请求
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);

        System.out.println("status: " + bulkResponse.status());
    }

    /**
     *  查询goods索引库中的数据
     */
    @Test
    public void testQueryBo() throws IOException {
        // 查询请求对象
        SearchRequest searcheRequest = new SearchRequest("goods");
        // 五大条件
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 添加查询条件
        sourceBuilder.query(QueryBuilders.matchQuery("title","数码"));
//
//        sourceBuilder.from();
//        // 显示数量
//        sourceBuilder.size();
//        // 高亮
//        sourceBuilder.highlighter();
//        // 排序
//        sourceBuilder.sort();
//        // 聚合
//        sourceBuilder.aggregation();
//        // 过滤查询字段
//        sourceBuilder.fetchSource()

        // 把查询条件添加到请求对象中
        searcheRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searcheRequest, RequestOptions.DEFAULT);

        SearchHits hits = searchResponse.getHits();

        float maxScore = hits.getMaxScore();
        // 命中个数
        long totalHits = hits.getTotalHits().value;
        System.out.println("totalHits = " + totalHits);

        SearchHit[] searchHits = hits.getHits();

        for (SearchHit searchHit : searchHits) {
            float score = searchHit.getScore();
            // 取出时将json转为goods
            Goods goods = JSON.parseObject(searchHit.getSourceAsString(), Goods.class);
            System.out.println("score 分数= " + score);
            System.out.println("goods 内容= " + goods);
        }
    }

    /**
     *     自动补全搜索
     * @throws IOException
     */
    @Test
    public void testSuggest() throws IOException {

        SearchRequest searchRequest = new SearchRequest("goods");

        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        SuggestBuilder suggestBuilder = new SuggestBuilder();
        // 添加自动补全条件 自定义名称electron-电子
        suggestBuilder.addSuggestion("electron", SuggestBuilders.completionSuggestion("name").prefix("s").size(40));

        // 将自动补全的条件加入构造工具中
        sourceBuilder.suggest(suggestBuilder);
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        // 根据key取到value得到对象
        // 继续获取对象中的自定义名称
        Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> electronSuggestion
                = searchResponse.getSuggest().getSuggestion("electron");

        electronSuggestion.forEach(suggest -> {
            // 通过Option取text的数据
            suggest.getOptions().forEach(Option -> {
                System.out.println("Option.getText() = " + Option.getText());
            });
        });
    }

    /**
     *  异步新增
     * @throws InterruptedException
     */
    @Test
        public void testAsyncAddDocument() throws InterruptedException {
            // 准备文档
            Goods goods = new Goods(5L, "松下电吹风", "松下电吹风 网红电吹风", 1599L);

            // 创建请求
            IndexRequest request = new IndexRequest("goods")
                    .id(goods.getId().toString())
                    .source(JSON.toJSONString(goods), XContentType.JSON);

            // 执行请求，第三个参数是回调处理
            client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                /**
                 * 执行成功时的回调，参数是IndexResponse结果
                 * @param indexResponse 执行结果
                 */
                @Override
                public void onResponse(IndexResponse indexResponse) {
                    System.out.println("我是成功的回调！" + indexResponse);
                }
                /**
                 * 执行失败时的回调，参数是异常信息
                 * @param e 异常信息
                 */
                @Override
                public void onFailure(Exception e) {
                    System.out.println("我是失败的回调！");
                    e.printStackTrace();
                }
            });

            System.out.println("我的异步方法调用完成~~");
            // 因为我们的程序结束会立即停止，接收不到回调结果，这里我们休眠一下，等待下回调结果
            Thread.sleep(2000L);
        }

    /**
     *   异步查询
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void testGetDocumentByIdAsync() throws IOException, InterruptedException {

        System.out.println("准备开始查询");
        // 准备一个查询文档的请求
        GetRequest request = new GetRequest("goods", "1");
        // 异步查询一个文档，耗时50ms
        client.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
            @Override
            public void onResponse(GetResponse response) {
                // 获取source
                String json = response.getSourceAsString();
                // 把json反序列化
                Goods goods = JSON.parseObject(json, Goods.class);

                System.out.println("查询结束，得到结果： " + goods);
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
            }
        });

        System.out.println("请求已经发出，等待执行结果！");

        Thread.sleep(2000);
    }

    /**
     *    Mono
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void testGetDocumentByIdAsyncMono() throws IOException, InterruptedException {

        System.out.println("准备开始查询");
        // 准备一个查询文档的请求
        GetRequest request = new GetRequest("goods", "1");

        // 将异步请求封装
        Mono<Goods> goodsMono = Mono.create(monoSink ->{

            client.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse response) {
                    // 获取source
                    String json = response.getSourceAsString();
                    // 把json反序列化
                    Goods goods = JSON.parseObject(json, Goods.class);

                    monoSink.success(goods);

                    System.out.println("查询结束，得到结果： " + goods);
                }
                @Override
                public void onFailure(Exception e) {
                    e.printStackTrace();
                }
            });
        });

        // 异步查询一个文档，耗时50ms
        System.out.println("请求已经发出，等待执行结果！");

        Thread.sleep(2000);
    }
}
